﻿
#include <stdlib.h>
#include <stdio.h>
#include <memory.h>
#include <assert.h>

#define LAND_BASE_CORE

#include "land_base.h"

#define DEFAULT_QUEUE_SIZE      32
#define SERVICE_WAITMS      5000
#define MSG_DATA_BUFF      0


typedef struct Service Service;
struct Service{
    LandSpinLock lock;
    uint32_t cap;
    uint32_t head;
    uint32_t tail;
    uint32_t queueCount;
    Service * next;
    int inGlobal;
    int isWorker; 
    int exit;
    LandMsg * queue;
    LandMsg cacheQueue[DEFAULT_QUEUE_SIZE];
    LandService h;
    void * ud;
    LandSession sessionIdx;
    LandMsgCallback msgCallback;
    LandExitSvrCallback exitCallback;
};


static struct{
    Service * head;
    Service * tail;
    LandSpinLock queueLock;
    LandStorage * storage;
    LandRWLock storageLock;
}ctx = {0};

static void
pushGlobalQueue (Service * s){
    assert(s->next == NULL);
    landLockSpin(&ctx.queueLock);
    if(ctx.tail) {
        ctx.tail->next = s;
        ctx.tail = s;
    } else {
        ctx.head = ctx.tail = s;
    }
    landUnlockSpin(&ctx.queueLock);
}


static Service *
popGlobalQueue (){
    landLockSpin(&ctx.queueLock);
    Service * s = ctx.head;
    if(s){
        ctx.head = s->next;
        if(!ctx.head){
            assert(s == ctx.tail);
            ctx.tail = NULL;
        }
        s->next = NULL;
    }
    landUnlockSpin(&(ctx.queueLock));
    return s;
}

static LandService
regInStorage (Service ** s){
    LandService h;
    landLockRWLockW(&ctx.storageLock);
    h = landRegInStorage(ctx.storage, (void **)s);
    landUnlockRWLockW(&ctx.storageLock);
    return h;
}

static Service *
findService (LandService h){
    Service * s;
    landLockRWLockR(&ctx.storageLock);
    s = landFindInStorage(ctx.storage, h);
    landUnlockRWLockR(&ctx.storageLock);
    return s;
}

static void
revServiceHandle (LandService h){
    landLockRWLockW(&ctx.storageLock);
    landRecInStorage(ctx.storage, h);
    landUnlockRWLockW(&ctx.storageLock);
}


static Service *
createService (Service* s) {
    landInitBaseSpin(&s->lock);
    s->cap = DEFAULT_QUEUE_SIZE;
    s->queue = malloc(sizeof(LandMsg) * s->cap);
    s->next = NULL;
    s->sessionIdx = 10000;
    s->isWorker = 0;
    s->queueCount = 0;
    return s;
}

static void
expandServiceQueue (Service * s, uint32_t cap){
    if (cap <= DEFAULT_QUEUE_SIZE) {
        return;
    }
    LandMsg* queue = malloc(sizeof(LandMsg) * cap);
    for (unsigned i = 0; i < s->cap; i++) {
        queue[i] = s->queue[(s->head + i) % s->cap];
    }

    if (s->cap > DEFAULT_QUEUE_SIZE) {
        free(s->queue);
    }
    s->head = 0;
    s->tail = s->cap;
    s->cap = cap;
    s->queue = queue;
}

static void
pushServiceQueue (Service* s, LandMsgSend* send){
    landLockSpin(&s->lock);
    s->queueCount++;
    LandMsg * msg = &(s->queue[s->tail]);
    msg->send = *send;
    if (++s->tail >= s->cap) {
        s->tail = 0;
    }
    if (s->head == s->tail) {
        expandServiceQueue(s, s->cap * 2);
    }
    if (s->isWorker && !s->inGlobal) {
        s->inGlobal = 1;
        pushGlobalQueue(s);
    }
    landUnlockSpin(&s->lock);
}

static int
popServiceQueue (Service* s, LandMsg* msg, int max) {
    int ret = 0;
    landLockSpin(&s->lock);
    while (s->head != s->tail && ret < max) {
        msg[ret] = s->queue[s->head++];
        s->queueCount --;
        ret += 1;
        int head = s->head;
        int tail = s->tail;
        int cap = s->cap;
        if (head >= cap) {
            s->head = head = 0;
        }
    }
    if(s->queueCount == 0 && s->cap > (DEFAULT_QUEUE_SIZE * 2)){
        free(s->queue);
        s->cap = DEFAULT_QUEUE_SIZE;
        s->queue = s->cacheQueue;
        s->head = s->tail = 0;
    }
    landUnlockSpin(&s->lock);
    return ret;
}


static void
freeServiceQueue (Service * s){
    if (s->exitCallback) {
        s->exitCallback(s->h, s->ud);
    }
    landClearSpin(&s->lock);
    LandMsg * msg;
    while (s->head != s->tail) {
        msg = &(s->queue[s->head++]);
        if(msg->send.data && msg->send.free){
            msg->send.free(msg->send.data);
        }
        if (s->head >= s->cap) {
            s->head = 0;
        }
    }
    free(s->queue);
    revServiceHandle(s->h);
}

void
dispatchMsg (Service * s, LandMsg * msg){
    s->msgCallback(msg);
    if (msg->send.data && msg->send.free) {
        msg->send.free(msg->send.data);
    }
}


static inline LandSession
incSession (Service * s){
    LandSession id = ++ s->sessionIdx;
    if(id == 0){
        s->sessionIdx ++;
        return 1;
    }
    return id;
}


int
landInitServiceContext (){
    landInitBaseSpin(&ctx.queueLock);
    landInitBaseRWLock(&ctx.storageLock);
    ctx.storage = landNewStorage(sizeof(Service), 0xFF + 1);
    return 0;
}

int
landExitServiceContext(){
    landClearSpin(&ctx.queueLock);
    landClearRWLock(&ctx.storageLock);
    return 0;
}

LAND_BASE_API LandService
landNewService (LandMsgCallback cb, void * ud){
    Service* s;
    LandService h = regInStorage(&s);
    createService(s);
    s->msgCallback = cb;
    s->ud = ud;
    s->h = h;
    LandMsgSend send = { 0 };
    send.ptype = LAND_PTYPE_STARTUP;
    pushServiceQueue(s, &send);
    return h;
}

int
landDispatchWorker (){
    Service* s = popGlobalQueue();
    if (!s){
        return 0;
    }
    LandMsg msg[8];
    int ret = popServiceQueue(s, msg, 8);
    if(!ret){
        return 0;
    }
    for (int i = 0; i < ret; i ++){
        dispatchMsg(s, &(msg[i]));
    }
    if (s->exit) {
        freeServiceQueue(s);
    }else {
        landLockSpin(&s->lock);
        if (s->tail != s->head) {
            s->inGlobal = 1;
        }else {
            s->inGlobal = 0;
        }
        if (s->inGlobal) {
            pushGlobalQueue(s);
        }
        landUnlockSpin(&s->lock);
    }
    return 1;
}


LAND_BASE_API int
landDispatchService (LandService h){
    Service* s = findService(h);
    if(!s){ return 1; }
    LandMsg msg[8];
    int ret = popServiceQueue(s, msg, 8);
    if(!ret){
        return 0;
    }
    for (int i = 0; i < ret; i ++){
        dispatchMsg(s, &(msg[i]));
    }
    return ret;
}

LAND_BASE_API int
landAttachWorker (LandService h){
    Service* s = findService(h);
    if(!s){ return 1; }
    if(s->isWorker){ return 1; }
    landLockSpin(&s->lock);
    s->isWorker = 1;
    if(!s->inGlobal && (s->tail != s->head)){
        s->inGlobal = 1;
        pushGlobalQueue(s);
    }
    landUnlockSpin(&s->lock);
    return 0;
}

LAND_BASE_API int
landSend (LandService h, LandMsgSend* send){
    Service* s = findService(h);
    if(!s){ return 1; }
    pushServiceQueue(s, send);
    return 0;
}

LAND_BASE_API int
landNewSession (LandService h, LandSession * session){
    Service* s = findService(h);
    if(!s){ return 1; }
    *session = incSession(s);
    return 0;
}

